/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package com.allyes.flume.source;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.flume.Event;
import org.apache.flume.event.EventBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Charsets;
import com.google.common.collect.Lists;
import com.google.common.io.ByteArrayDataOutput;
import com.google.common.io.ByteStreams;

public class TailFile {
  private static final Logger logger = LoggerFactory.getLogger(TailFile.class);

  private static final String LINE_SEP = "\n";
  private static final int LOGFILE_LINE_SIZE_INIT = 1024;

  private final String path;
  private final String filename;
  private final long inode;
  private long lastUpdated;
  private boolean needTail;
  private final Map<String, String> staticHeaders;
  private final HashSet<String> dynamicHeaders;
  private RandomAccessFile raf;

  // left : offset of the file, start from 0
  // right: the line index where 'offset' locate, start from 1
  private final Pair<Long, Integer> commitedPosAndLineIndex = new Pair<Long, Integer>();
  private final Pair<Long, Integer> nextPosAndLineIndex = new Pair<Long, Integer>();


  public TailFile(
      File file,
      long inode,
      long pos,
      int lineIndex,
      Map<String, String> staticHeaders,
      HashSet<String> dynamicHeaders) throws IOException {

    this.path = file.getAbsolutePath();
    this.filename = file.getName();
    this.inode = inode;
    this.staticHeaders = staticHeaders;
    this.dynamicHeaders = dynamicHeaders;
    this.lastUpdated = 0L;
    this.needTail = true;

    this.raf = new RandomAccessFile(file, "r");
    updatePos(pos, lineIndex);
  }

  public RandomAccessFile getRaf() { return raf; }
  public String getPath() { return path; }
  public long getInode() { return inode; }
  public long getLastUpdated() { return lastUpdated; }
  public boolean needTail() { return needTail; }

  public Pair<Long, Integer> getCommitedPosAndLineIndex() {
    return commitedPosAndLineIndex;
  }

  public void setLastUpdated(long lastUpdated) {
    this.lastUpdated = lastUpdated;
  }

  public void setNeedTail(boolean needTail) {
    this.needTail = needTail;
  }

  public void commit(long updateTime) {
    commitedPosAndLineIndex.set(nextPosAndLineIndex);
    setLastUpdated(updateTime);
  }

  public void rollBack() throws IOException {
    raf.seek(this.commitedPosAndLineIndex.left);
    nextPosAndLineIndex.set(commitedPosAndLineIndex);
  }

  private void updatePos(long pos, int lineIndex) throws IOException {
    raf.seek(pos);
    if (lineIndex < 1) {
      lineIndex = getLineIndexAtPos(raf, pos);
    }
    nextPosAndLineIndex.set(pos, lineIndex);
    commitedPosAndLineIndex.set(nextPosAndLineIndex);
  }

  public boolean updatePos(String path, long inode, long pos, int lineIndex) throws IOException {
    if (this.inode == inode && this.path.equals(path)) {
      updatePos(pos, lineIndex);
      logger.info("Updated position, file: " + path + ", inode: " + inode +
          ", pos: " + pos + ", line index: " + lineIndex);
      return true;
    }
    return false;
  }

  public List<Event> readEvents(
      int numEvents,
      boolean backoffWithoutNL) throws IOException {

    List<Event> events = Lists.newLinkedList();

    while (events.size() < numEvents) {
      Event event = readEvent(backoffWithoutNL);
      if (null == event) {
        break;
      }
      events.add(event);
    }

    return events;
  }

  // return null means no data to read in this file now
  private Event readEvent(boolean backoffWithoutNL) throws IOException {
    final Long posBefore = raf.getFilePointer();
    String line = readLine();

    if (null == line) {
      return null;
    }

    if (backoffWithoutNL && !line.endsWith(LINE_SEP)) {
      logger.info("Backing off in file without newline: "
          + path + ", inode: " + inode + ", pos: " + raf.getFilePointer());
      raf.seek(posBefore);
      return null;
    }

    // get the line index just read
    final int lineIndex;
    if (posBefore.equals(this.nextPosAndLineIndex.left)) {
      lineIndex = this.nextPosAndLineIndex.right;
    } else {
      lineIndex = getLineIndexAtPos(raf, posBefore);
    }
    nextPosAndLineIndex.set(raf.getFilePointer(), lineIndex + 1);

    line = StringUtils.remove(line, LINE_SEP);
    byte[] eventBody = line.getBytes(Charsets.UTF_8);
    Event event = EventBuilder.withBody(eventBody);
    setStaticHeaders(event);
    setDynamicHeaders(event, posBefore, lineIndex);
    return event;
  }

  private void setStaticHeaders(Event event) {
    if (staticHeaders != null) {
      event.getHeaders().putAll(staticHeaders);
    }
  }

  private void setDynamicHeaders(Event event, Long offset, int lineIndex) {
    if (dynamicHeaders == null) {
      return;
    }

    for (String header: dynamicHeaders) {
      switch (header) {
        case TaildirSourceConfigurationConstants.TXT_INODE:
          event.getHeaders().put(header, Long.toString(this.inode));
          break;
        case TaildirSourceConfigurationConstants.TXT_PATH:
          event.getHeaders().put(header, this.path);
          break;
        case TaildirSourceConfigurationConstants.TXT_FILE_NAME:
          event.getHeaders().put(header, this.filename);
          break;
        case TaildirSourceConfigurationConstants.TXT_LINE_INDEX:
          event.getHeaders().put(header, Integer.toString(lineIndex));
          break;
        case TaildirSourceConfigurationConstants.TXT_BYTE_OFFSET:
          event.getHeaders().put(header, offset.toString());
          break;
      }
    }
  }

  private String readLine() throws IOException {
    ByteArrayDataOutput out = ByteStreams.newDataOutput(LOGFILE_LINE_SIZE_INIT);

    int i = 0;
    int c;
    while ((c = raf.read()) != -1) {
      i++;
      out.write((byte) c);
      if (c == LINE_SEP.charAt(0)) {
        break;
      }
    }

    if (i == 0) {
      return null;
    }

    String rs = new String(out.toByteArray(), Charsets.UTF_8);
    return rs;
  }

  // Line index starts from 1.
  // Return 0 if 'position' is invalid or beyond the file size.
  // Note:
  //   We should resume the inner offset of 'raf' before returning.
  //   If 'position' point to the trailing ending char of some line, then return
  //     the line index of that line.
  public static int getLineIndexAtPos(RandomAccessFile raf, final long position)
      throws IOException {
    if (position < 0 || position > raf.length()) {
      return 0;
    }
    if (position == 0) {
      return 1;
    }

    final long posBefore = raf.getFilePointer();
    raf.seek(0);

    int lineNum = 1;
    final char END_CHAR = LINE_SEP.charAt(0);
    boolean lastCharIsEndChar = false;
    for (long i = 0; i <= position; i++) {
      final int c = raf.read();
      if (lastCharIsEndChar) {
        lineNum++;
      }
      lastCharIsEndChar = (c == END_CHAR);

      if (c == -1) {
        break;
      }
    }

    raf.seek(posBefore);
    return lineNum;
  }

  public static int getCurrentLineNum(RandomAccessFile raf) throws IOException {
    return getLineIndexAtPos(raf, raf.getFilePointer());
  }

  public void close() {
    try {
      if (raf != null) {
        raf.close();
        raf = null;
        long now = System.currentTimeMillis();
        setLastUpdated(now);
      }
    } catch (IOException e) {
      logger.error("Failed closing file: " + path + ", inode: " + inode, e);
    }
  }

  public static class CompareByLastModifiedTime implements Comparator<File> {
    @Override
    public int compare(File f1, File f2) {
      return Long.valueOf(f1.lastModified()).compareTo(f2.lastModified());
    }
  }
}
